/**
 * Copyright 2007-2008. Chongqing First Information & Network Co., Ltd. All
 * rights reserved. <a>http://www.cqfirst.com.cn</a>
 */
package com.cqcis.uip.base.framework.base.adaptor;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ThreadPoolExecutor;

import org.apache.mina.common.ExecutorThreadModel;
import org.apache.mina.common.IoFilter;
import org.apache.mina.common.IoSession;
import org.apache.mina.filter.codec.ProtocolCodecFilter;

import com.cqcis.uip.base.framework.codec.ObjectDecoder;
import com.cqcis.uip.base.framework.codec.ObjectEncoder;
import com.cqcis.uip.base.framework.codec.impl.DefaultProtocolObjectDecoder;
import com.cqcis.uip.base.framework.codec.impl.DefaultProtocolObjectEncoder;
import com.cqcis.uip.base.framework.common.Task;
import com.cqcis.uip.base.framework.core.RegisteredServiceContext;
import com.cqcis.uip.base.framework.core.TaskResultProcessor;
import com.cqcis.uip.base.framework.core.ThreadPoolConfig;
import com.cqcis.uip.base.framework.filters.Filter;
import com.cqcis.uip.base.framework.filters.impl.DefaultProtocolReadFilter;
import com.cqcis.uip.base.framework.filters.impl.DefaultProtocolWriteFilter;
import com.cqcis.uip.base.framework.service.InterfaceLogService;
import com.cqcis.uip.base.framework.service.StreamNameUtil;
import com.cqcis.uip.base.framework.service.StreamOutput;
import com.cqcis.uip.base.framework.utils.ApplicationExecutorFilter;
import com.cqcis.uip.base.framework.utils.LogServiceThreadPoolExecutor;

/**
 * socket类型注册服务的抽象实现
 * 
 * @author huangmy
 * @date   2007-12-17
 */
public abstract class AbstractSocketService extends AbstractRegisteredService {
	private ObjectDecoder decoder;      /*定义写数据*/
	private ObjectEncoder encoder;      /*定义读数据*/
	private List<IoFilter> readFilters = new ArrayList<IoFilter>();      /*处理接收到的数据*/
	private List<IoFilter> writeFilters = new ArrayList<IoFilter>(); 	  /*处理要发送的数据*/
	
	private StreamNameUtil streamNameUtil;  // 字节流文件名服务
    private StreamOutput   streamOutput;  // 字节流读写服务
	
	protected int threadPoolCoreSize = 4;
	protected int threadPoolMaxSize = 16;
	protected int threadPoolKeepAliveTime = 60;
	private ThreadPoolConfig threadPoolConfig; // 线程池配置
	private ThreadPoolExecutor socketThreadModelExecutor; // 过滤器分派线程池
	private ThreadPoolExecutor applicationFilterExecutor; // 业务层异步通知线程池
	private InterfaceLogService interfaceLogService;
		
	private int port;	                  /*关注端口*/
	private int clientNumberMax = 1;          /*允许的最大客户端*/
	private boolean keepAlive = true;      // 是否长连接
	
	private boolean streamRecord = true; // 是否记录字节流
	
	public ThreadPoolConfig getDefaultThreadPoolConfig() {		
		return this.threadPoolConfig;
	}
	
	/**
	 * 字节流记录
	 * 
	 * @return
	 */
	protected boolean isStreamRecord() {
		return this.streamRecord;
	}
	
	/**
	 * 忽略Task的处理结果，对Task处理的监控采用查看Task状态的方式进行;
	 * 或者在handler层实现{@link TaskResultProcessor}接口
	 */
	@Override
	public boolean processResult(Task result) {		
		return true;
	}
		
	/**
	 * 设置线程模型配置
	 * 
	 * @param etm
	 */
	protected abstract void setThreadPoolExecutorModelConfig(ExecutorThreadModel etm);
	
	/**
	 * 当前sessions
	 * 
	 * @return
	 */
	public abstract Collection<IoSession> getManagedSessions();
	
	/**
	 * 获取bundleId
	 * 
	 * @param serviceContext
	 * @return
	 */
	protected long getBundleId(RegisteredServiceContext serviceContext) {
		return serviceContext.getRegisteredService().getBundleContext().getBundle().getBundleId();
	}

	public void start(RegisteredServiceContext serviceContext) throws Exception {		
		super.start(serviceContext);
		
		this.threadPoolConfig = new ThreadPoolConfig(this.threadPoolCoreSize, this.threadPoolMaxSize, this.threadPoolKeepAliveTime);
		
		DefaultProtocolObjectDecoder decodec = new DefaultProtocolObjectDecoder();
		decodec.setDecoder(decoder);
		
		DefaultProtocolObjectEncoder encodec = new DefaultProtocolObjectEncoder();
		encodec.setEncoder(encoder);
		
		encodec.registerStreamNameUtil(streamNameUtil);
		encodec.registerStreamOutput(streamOutput);
		
		decodec.registerStreamNameUtil(streamNameUtil);
		decodec.registerStreamOutput(streamOutput);
		
		ExecutorThreadModel etm = ExecutorThreadModel.getInstance(getInterfaceName() 
				+ getBundleId(serviceContext)); // 默认线程模型
		
		this.socketThreadModelExecutor = LogServiceThreadPoolExecutor.getLogExecutor(threadPoolCoreSize, 
				threadPoolMaxSize, threadPoolKeepAliveTime, serviceContext.getLogService(), getLogConfiguration(), "(F)"+getInterfaceName());
		
		etm.setExecutor(this.socketThreadModelExecutor);
		
		// 设置
		setThreadPoolExecutorModelConfig(etm);
		
		/*编解码*/
		addFilter("ObjectReadWriteFilter", new ProtocolCodecFilter(encodec, decodec));	
		
		int i = 0;
		for (Iterator<IoFilter> it = readFilters.iterator(); it.hasNext(); i ++) {
			IoFilter readFilter = it.next();
			
			addFilter("readFilter" + i, readFilter);
		}
	    
		i = 0;
		for (Iterator<IoFilter> it = writeFilters.iterator(); it.hasNext(); i ++) {
			IoFilter writeFilter = it.next();
			
			addFilter("writeFilter" + i, writeFilter);
		}

		/*如果业务层线程存在阻塞, 则有必要在业务层增加一个线程池, 否则无法实现事件异步通知*/	
		this.applicationFilterExecutor = LogServiceThreadPoolExecutor.getLogExecutor(threadPoolCoreSize, threadPoolMaxSize, 
				threadPoolKeepAliveTime, serviceContext.getLogService(), 
				getLogConfiguration(), "(H)"+getInterfaceName());
		ApplicationExecutorFilter executorFilter = new ApplicationExecutorFilter(this.applicationFilterExecutor, "ThreadPoolExecutorForAppalication");
		
		/*增加线程池*/		
		addFilter(executorFilter.getName(), executorFilter);
	}
	
	public void stop(RegisteredServiceContext serviceContext) throws Exception {
		if (this.socketThreadModelExecutor != null) {
			this.socketThreadModelExecutor.shutdownNow();
		}
		
		if (this.applicationFilterExecutor != null) {
			this.applicationFilterExecutor.shutdownNow();
		}
		
		super.stop(serviceContext);
	}
	
	/**
	 * 获取所有的filters
	 * 
	 * @return
	 */
	public abstract Map<String, IoFilter> getFilters();
	
	/**
	 * 增加过滤器
	 * @param filterName
	 * @param filter
	 */
	protected abstract void addFilter(String filterName, IoFilter filter);
	
	/**
	 * 读过滤器
	 * 
	 * @param readFilters
	 */
	public void setReadFilters(List<Filter> readFilters) {
		List<IoFilter> filters = new ArrayList<IoFilter>();
		
		for (Iterator<Filter> it = readFilters.iterator(); it.hasNext();) {
			Filter filter = it.next();
			
			DefaultProtocolReadFilter readFilter = new DefaultProtocolReadFilter();
			readFilter.setFilter(filter);
			
			filters.add(readFilter);
		}
		
		this.readFilters = filters;
	}
    
	/**
	 * 写过滤器
	 * 
	 * @param writeFilters
	 */
	public void setWriteFilters(List<Filter> writeFilters) {
        List<IoFilter> filters = new ArrayList<IoFilter>();
		
		for (Iterator<Filter> it = writeFilters.iterator(); it.hasNext();) {
			Filter filter = it.next();
			
			DefaultProtocolWriteFilter writeFilter = new DefaultProtocolWriteFilter();
			writeFilter.setFilter(filter);
			
			filters.add(writeFilter);
		}
		
		this.writeFilters = filters;
	}

	public void setPort(int port) {
		this.port = port;
	}

	public void setClientNumberMax(int clientNumberMax) {
		this.clientNumberMax = clientNumberMax;
	}
    
	public void setDecoder(ObjectDecoder decoder) {
		this.decoder = decoder;
	}

	public void setEncoder(ObjectEncoder encoder) {
		this.encoder = encoder;
	}

	protected int getPort() {
		return port;
	}

	public int getClientNumberMax() {
		return clientNumberMax;
	}

	public void setThreadPoolCoreSize(int threadPoolCoreSize) {
		if (threadPoolCoreSize <= 0) {
			return;
		}
		
		this.threadPoolCoreSize = threadPoolCoreSize;
	}

	public void setThreadPoolMaxSize(int threadPoolMaxSize) {
		if (threadPoolMaxSize <= 0) {
			return;
		}
		
		this.threadPoolMaxSize = threadPoolMaxSize;
	}

	public void setThreadPoolKeepAliveTime(int threadPoolKeepAliveTime) {
		if (threadPoolKeepAliveTime <= 0) {
			return;
		}
		
		this.threadPoolKeepAliveTime = threadPoolKeepAliveTime;
	}

	public void setStreamNameUtil(StreamNameUtil streamNameUtil) {
		this.streamNameUtil = streamNameUtil;
	}

	public void setStreamOutput(StreamOutput streamOutput) {
		this.streamOutput = streamOutput;
	}
	
	public void setInterfaceLogService(InterfaceLogService interfaceLogService) {
		this.interfaceLogService = interfaceLogService;
	}
	
	protected InterfaceLogService getInterfaceLogService() {
		return this.interfaceLogService;
	}
	
	public boolean isKeepAlive() {
		return this.keepAlive;
	}
	
	public void setKeepAlive(boolean keepAlive) {
		this.keepAlive = keepAlive;
	}

	public void setStreamRecord(boolean streamRecord) {
		this.streamRecord = streamRecord;
	}
}
